package com.sskj.common.http;

/*
 * 作者 :吕志豪
 * 简书：https://www.jianshu.com/u/6e525b929aac
 * github：https://github.com/lvzhihao100
 * 描述：
 * 创建时间：2018-07-26 11:31
 */

import com.sskj.common.R;
import com.sskj.common.base.App;

import org.reactivestreams.Subscriber;

import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;
import javax.xml.transform.Source;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.Observer;
import io.reactivex.android.MainThreadDisposable;
import io.reactivex.schedulers.Schedulers;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;


public abstract class WebSocketObserver extends Flowable<String> {
    private String wsUrl;
    private OkHttpClient okHttpClient;
    private Request request;
    private Subscriber<? super String> observer;
    private boolean isClose;
    private WebSocket mWebSocket;

    public WebSocketObserver(String wsUrl) {
        this.wsUrl = wsUrl;

    }

    public abstract void onOpen(WebSocket webSocket, Response response);

    @Override
    protected void subscribeActual(Subscriber<? super String> subscriber) {
        this.observer = subscriber;
        connect();

    }

    private void connect() {
        isClose = false;
        okHttpClient = new OkHttpClient.Builder()
                .readTimeout(13, TimeUnit.SECONDS).build();
        request = new Request.Builder()
                .url(wsUrl)
                .build();
        mWebSocket = okHttpClient.newWebSocket(request, new WebSocketListener() {
            @Override
            public void onOpen(WebSocket webSocket, Response response) {
                super.onOpen(webSocket, response);
                System.out.println(App.INSTANCE.getString(R.string.commonJavaWebSocketObserver1));
                WebSocketObserver.this.onOpen(webSocket, response);
            }

            @Override
            public void onMessage(WebSocket webSocket, String text) {
                super.onMessage(webSocket, text);
                observer.onNext(text);
            }

            @Override
            public void onClosing(WebSocket webSocket, int code, String reason) {
                super.onClosing(webSocket, code, reason);
                webSocket.close(1001, App.INSTANCE.getString(R.string.commonJavaWebSocketObserver2));
                System.out.println(App.INSTANCE.getString(R.string.commonJavaWebSocketObserver3) + "onClosing" + reason);
                if (!isClose) {
                    reConnect();
                }
            }

            @Override
            public void onFailure(WebSocket webSocket, Throwable t, @Nullable Response response) {
                super.onFailure(webSocket, t, response);
//                    webSocket.close(1001, App.INSTANCE.getString(R.string.commonJavaWebSocketObserver2));
                System.out.println(App.INSTANCE.getString(R.string.commonJavaWebSocketObserver3) + "onFailure" + t.toString());
                if (!isClose) {

                    reConnect();
                }

            }

            private void reConnect() {
                System.out.println("15秒后重新连接。。。");

                Flowable.timer(15, TimeUnit.SECONDS)
                        .subscribeOn(Schedulers.newThread())
                        .subscribe(i -> {
                            System.out.println("重新连接中。。。");
                            connect();
                        }, e -> {
                            System.out.println("15秒连接出错" + e.toString());
                        });
            }

            @Override
            public void onClosed(WebSocket webSocket, int code, String reason) {
                super.onClosed(webSocket, code, reason);
                System.out.println(App.INSTANCE.getString(R.string.commonJavaWebSocketObserver6) + reason);
                isClose = true;
                if (code != 1000 && !App.INSTANCE.getString(R.string.commonJavaWebSocketObserver7).equals(reason)) {
                    reConnect();
                }
            }
        });
    }

    public void send(String msg) {
        if (mWebSocket != null) {
            mWebSocket.send(msg);
        }
    }

    public void close() {
        if (mWebSocket != null) {
            isClose = true;
            System.out.println("onDispose 关闭了");
            mWebSocket.close(1000, App.INSTANCE.getString(R.string.commonJavaWebSocketObserver7));
            mWebSocket = null;
        }
    }


}